package com.qupeng.demo.kafka.kafkaspringbootproducer.admin;

import com.qupeng.demo.kafka.kafkaspringbootproducer.admin.model.Topic;
import com.qupeng.demo.kafka.kafkaspringbootproducer.model.CommonResponse;
import com.qupeng.demo.kafka.kafkaspringbootproducer.config.KafkaProperties;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.TopicCollection;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.KafkaException;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.web.bind.annotation.*;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

@RestController
@RequestMapping("/admin/sync/topics")
public class TopicSyncAdminController {

    private static final Logger LOGGER = LoggerFactory.getLogger(TopicSyncAdminController.class);

    @Autowired
    private KafkaAdmin kafkaAdmin;

    @Autowired
    private AdminClient adminClient;

    @Autowired
    private KafkaProperties kafkaProperties;

    @GetMapping("")
    public CommonResponse<List<String>> topics() {
        ListTopicsResult result = adminClient.listTopics(new ListTopicsOptions().listInternal(false).timeoutMs(3000));

        try {
            List<String> topicListings = result.listings().get().stream().map(String::valueOf).collect(Collectors.toList());
            return new CommonResponse<>("kafka-admin", topicListings);
        } catch (InterruptedException e) {
            return new CommonResponse<>("001", e.getMessage(), "kafka-admin");
        } catch (ExecutionException e) {
            return new CommonResponse<>("002", e.getMessage(), "kafka-admin");
        }

    }

    @GetMapping("/{name}")
    public CommonResponse<Topic> describeTopic(@PathVariable String name) {
        try {
            Map<String, TopicDescription> topicDescriptionMap = kafkaAdmin.describeTopics(name);
            TopicDescription topicDescription = topicDescriptionMap.get(name);
            return new CommonResponse<>("kafka-admin", Util.convert(topicDescription));
        } catch (KafkaException ex) {
            if (ex.getCause() instanceof UnknownTopicOrPartitionException) {
                //TODO: maybe create a new one.
            }
            return new CommonResponse<>("006", ex.getMessage(), "kafka-admin");
        }
    }

    @PostMapping("")
    public CommonResponse<Topic> createTopic(@RequestBody Topic topic) {

        CreateTopicsResult newTopicResult = adminClient.createTopics(Collections.singletonList(new NewTopic(topic.getName(), topic.getPartitionCount(), topic.getReplicaFactor())));
        try {
            Uuid topicId = newTopicResult.topicId(topic.getName()).get();
            TopicDescription topicDescription = adminClient.describeTopics(Collections.singletonList(topic.getName()), new DescribeTopicsOptions().timeoutMs(5000)).topicNameValues().get(topic.getName()).get();
            return new CommonResponse<>("kafka-admin", Util.convert(topicDescription));
        } catch (InterruptedException | ExecutionException ex) {
            return new CommonResponse<>("007", ex.getMessage(), "kafka-admin");
        }

    }

    @DeleteMapping("/{name}")
    public CommonResponse<List<String>> deleteTopic(@PathVariable String name) {
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(TopicCollection.ofTopicNames(Collections.singletonList(name)));
        try {
            deleteTopicsResult.all().get();
            ListTopicsResult result = adminClient.listTopics(new ListTopicsOptions().listInternal(false).timeoutMs(3000));
            List<String> topicListings = result.listings().get().stream().map(String::valueOf).collect(Collectors.toList());
            return new CommonResponse<>("kafka-admin", topicListings);
        } catch (InterruptedException | ExecutionException ex) {
            return new CommonResponse<>("008", ex.getMessage(), "kafka-admin");
        }
    }

    @PutMapping("/{name}")
    public CommonResponse<Topic> modifyTopic(@PathVariable String name, @RequestBody Topic topic) {
        try {
            NewTopic modifiedTopic = new NewTopic(name, topic.getPartitionCount(), topic.getReplicaFactor());
            kafkaAdmin.createOrModifyTopics(modifiedTopic);
            Map<String, TopicDescription> topicDescriptionMap = kafkaAdmin.describeTopics(name);
            TopicDescription topicDescription = topicDescriptionMap.get(name);
            return new CommonResponse<>("kafka-admin", Util.convert(topicDescription));
        } catch (KafkaException ex) {
            return new CommonResponse<>("009", ex.getMessage(), "kafka-admin");
        }
    }
}
